package com.atguigu.realtime.util;

import com.atguigu.realtime.bean.VisitorStats;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.lang.reflect.Modifier;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/2/9 15:36
 */
public class MySinkUtil {
    
    /**
     * 到一个SinkFunction  sql插入语句只能是这样的:
     * insert into tableName(f1,f2,...)values(v1,v2,...)
     *
     * @param tableName
     * @param <T>
     * @return
     */
    public static <T> SinkFunction<T> getClickHouseSink(String db, String tableName, Class<T> tClass) {
        // String insertSql = "insert into visitor_stats_2021(f1,f2,....)values(?,?,?,?,?,?,?,?,?,?,?,?)";
        String clickHouseUrl = "jdbc:clickhouse://hadoop162:8123/" + db;
        String clickHourDriverName = "ru.yandex.clickhouse.ClickHouseDriver";
        
        // 根据表名拼接sql语句
        final Field[] fields = tClass.getDeclaredFields();
        final StringBuilder sql = new StringBuilder();
        sql.append("insert into ").append(tableName).append(" (");
        for (Field field : fields) {
            // 可能有些属性不需要写出去, 这样的字段我们用transient标识
            if (!Modifier.isTransient(field.getModifiers())) {
                final String fieldName = field.getName();
                sql.append(fieldName).append(",");
            }
        }
        sql.deleteCharAt(sql.length() - 1); // 拼接的最后一个逗号去掉
        sql.append(")values(");
        for (Field field : fields) {
            if (!Modifier.isTransient(field.getModifiers())) {
                sql.append("?,");
            }
        }
        sql.deleteCharAt(sql.length() - 1); // 拼接的最后一个逗号去掉
        sql.append(")");
        return getJdbcSink(clickHouseUrl,
                           clickHourDriverName,
                           sql.toString());
    }
    
    private static <T> SinkFunction<T> getJdbcSink(String url,
                                                   String driver,
                                                   String sql) {
        System.out.println(sql);
        return JdbcSink.<T>sink(sql,
                                (ps, t) -> {
                                    try {
                                        final Field[] fields = t.getClass().getDeclaredFields();
                                        for (int i = 0, position = 1; i < fields.length; i++) {
                    
                                            if (!Modifier.isTransient(fields[i].getModifiers())) {
                                                fields[i].setAccessible(true);
                                                Object v = fields[i].get(t);
                                                ps.setObject(position++, v);
                                            }
                                        }
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
            
                                },
                                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                    .withUrl(url)
                                    .withDriverName(driver)
                                    .build()
        );
        
    }
    
    public static void main(String[] args) throws NoSuchFieldException {
        final Field stt = VisitorStats.class.getDeclaredField("edt");
        final int mod = stt.getModifiers();
        System.out.println(Modifier.isTransient(mod));
    }
}
